package com.navinfo.tripanalysis.offline.service.impl.huashen;

import com.navinfo.tripanalysis.common.pojo.RunningType;
import com.navinfo.tripanalysis.common.pojo.RunningTypeEnum;
import com.navinfo.tripanalysis.common.vehicle.VehicleInfo;
import com.navinfo.tripanalysis.offline.service.LoadVehicleInfoService;
import com.navinfo.tripanalysis.offline.service.impl.AbstractLoadVehicleInfoService;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import scala.Tuple2;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.stream.DoubleStream;

/**
 * 一汽加载车辆的配置信息
 * @author 沈东生
 */
@RunningType(value = RunningTypeEnum.HS, interfaceClass =LoadVehicleInfoService.class )
public class LoadVehicleInfoServiceImpl extends AbstractLoadVehicleInfoService {

    @Override
    protected String getTable() {
        String sql = "(\tSELECT \n" +
                "    v.terminal_id terminal_id,\n" +
                "    ra.speed_ratio rearaxle_ratio,\n" +
                "    gb.speed_ratios gear_ratios,\n" +
                "    wl.radius wheel_radius,\n" +
                "    ng.eco_rotation eco_rotation,\n" +
                "    ng.max_torque  max_torque,\n" +
                "    weight.no wloadno,\n" +
                "    weight.empty_load,\n" +
                "    weight.half_load,\n" +
                "    weight.full_load\n" +
                "FROM tr_vehicle v\n" +
                "   LEFT JOIN tr_vehicle_gearbox gb ON v.gearbox_no = gb.no\n" +
                "   LEFT JOIN tr_vehicle_rearaxle ra ON v.rearaxle_no = ra.no\n" +
                "   LEFT JOIN tr_vehicle_wheel wl ON v.wheel_no = wl.no\n" +
                "   LEFT JOIN tr_vehicle_engine ng ON v.engine_no = ng.no\n" +
                "   LEFT JOIN tr_vehicle_weightload weight ON v.weightload_no = weight.no) t";

        logger.info("加载车辆配置信息的sql：{}", sql);
        return sql;
    }

    @Override
    protected JavaPairRDD<Long, VehicleInfo> getVehicleInfoPair(SparkSession session, JavaRDD<Row> rowJavaRDD) {
        logger.info("加载车辆配置信息...");
        return rowJavaRDD.mapPartitionsToPair((PairFlatMapFunction<Iterator<Row>, Long, VehicleInfo>) LoadVehicleInfoServiceImpl::getVehicleInfoPair);
    }

    private static Iterator<Tuple2<Long, VehicleInfo>> getVehicleInfoPair(Iterator<Row> iterator) {
        List<Tuple2<Long, VehicleInfo>> vehicleInfos = new ArrayList<>();
        while (iterator.hasNext()) {
            Row row = iterator.next();
            VehicleInfo vehicleInfo = new VehicleInfo();

            //通讯号
            long tid = row.getLong(0);
            vehicleInfo.setTerminalId(tid);

            //后桥速比
            String rearAxleRatioStr = row.getString(1);
            vehicleInfo.setRearAxleSpeedRatio(StringUtils.isEmpty(rearAxleRatioStr) ? 0 : Double.parseDouble(rearAxleRatioStr));

            //各档位速比
            String gearRatios = row.getString(2);
            if (StringUtils.isNotEmpty(gearRatios)) {
                String[] split = gearRatios.split(",");
                DoubleStream doubleStream = Arrays.stream(split).mapToDouble(Double::valueOf);
                vehicleInfo.setGearsSpeedRatio(doubleStream.toArray());
            }

            //车轮半径
            String wheelRadiusStr = row.getString(3);
            vehicleInfo.setWheelRadius(StringUtils.isEmpty(wheelRadiusStr) ? 0 : Double.parseDouble(wheelRadiusStr));

            //经济转速
            String ecoRotationStr = row.getString(4);
            if (StringUtils.contains(ecoRotationStr, "-")) {
                String[] ss = ecoRotationStr.split("-");
                vehicleInfo.setEngineECORotationLow(Double.parseDouble(ss[0]));
                vehicleInfo.setEngineECORotationHigh(Double.parseDouble(ss[1]));
            }

            //最大输出扭矩
            String maxTorqueStr = row.getString(5);
            vehicleInfo.setEngineMaxTorque(StringUtils.isEmpty(maxTorqueStr) ? 0 : Double.parseDouble(maxTorqueStr));

            //载重信息
            vehicleInfo.setWloadNo(row.getString(6));
            vehicleInfo.setWloadEmpty(getInteger(row,"empty_load",null));
            vehicleInfo.setWloadHalf(getInteger(row,"half_load",null));
            vehicleInfo.setWloadFull(getInteger(row,"full_load",null));


            //经济车速赋值
            double wheelRadius = vehicleInfo.getWheelRadius();
            double rearAxleSpeedRatio = vehicleInfo.getRearAxleSpeedRatio();
            double engineECORotationLow = vehicleInfo.getEngineECORotationLow();
            double engineECORotationHigh = vehicleInfo.getEngineECORotationHigh();
            double[] gearsSpeedRatio = vehicleInfo.getGearsSpeedRatio();
            double highGearSpeedRatio = 0;
            if(null!=gearsSpeedRatio && gearsSpeedRatio.length>0){
                highGearSpeedRatio =  gearsSpeedRatio[gearsSpeedRatio.length-1];
            }
            //0.377 * 车轮半径(米) * 经济转速（RPM） / 后桥速比 / 最高档速比
            if(wheelRadius>0 && rearAxleSpeedRatio>0 && engineECORotationLow>0 && engineECORotationHigh>0 && highGearSpeedRatio>0 ){
                vehicleInfo.setEconomicSpeedLow(BigDecimal.valueOf(0.377 * wheelRadius * engineECORotationLow / rearAxleSpeedRatio / highGearSpeedRatio).setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue());
                vehicleInfo.setEconomicSpeedHigh(BigDecimal.valueOf(0.377 * wheelRadius * engineECORotationHigh / rearAxleSpeedRatio / highGearSpeedRatio).setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue());
            }

            vehicleInfos.add(new Tuple2<>(tid, vehicleInfo));
        }
        return vehicleInfos.iterator();
    }

    private static Integer getInteger(Row rs,String colName,Integer defaultValue)  {
        Object obj = rs.getAs(colName);
        if(null != obj){
            try {
                return Integer.valueOf(obj.toString());
            }catch (Exception ex){
                ex.printStackTrace();
                return defaultValue;
            }
        }
        return defaultValue;
    }
}

